package com.alibaba.dubbo.rpc.protocol.http;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;

import javax.servlet.ServletException;
import javax.servlet.ServletInputStream;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.springframework.web.HttpRequestHandler;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.HeaderProvider;
import com.alibaba.dubbo.common.io.Bytes;
import com.alibaba.dubbo.common.io.UnsafeByteArrayOutputStream;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.serialize.ObjectInput;
import com.alibaba.dubbo.common.serialize.ObjectOutput;
import com.alibaba.dubbo.common.serialize.Serialization;
import com.alibaba.dubbo.common.utils.ReflectUtils;
import com.alibaba.dubbo.common.utils.StringUtils;
import com.alibaba.dubbo.remoting.Interceptor;
import com.alibaba.dubbo.remoting.MessageCodec;
import com.alibaba.dubbo.remoting.MessageDecode;
import com.alibaba.dubbo.remoting.MessageEncode;
import com.alibaba.dubbo.remoting.RemoteCalling;
import com.alibaba.dubbo.remoting.interceptor.InterceptorUtils;
import com.alibaba.dubbo.remoting.msgcodec.BridgeDecodeInputStream;
import com.alibaba.dubbo.remoting.msgcodec.BridgeEncodeOutputStream;
import com.alibaba.dubbo.remoting.msgcodec.DecodeWrapper;
import com.alibaba.dubbo.remoting.msgcodec.EncodeWrapper;
import com.alibaba.dubbo.remoting.msgcodec.MessageCodecSupport;
import com.alibaba.dubbo.remoting.transport.CodecSupport;
import com.alibaba.dubbo.rpc.RpcContext;

public class DubboServletHandler implements HttpRequestHandler {
	private static final Logger log = LoggerFactory
			.getLogger(DubboServletHandler.class);
	public static final byte RESPONSE_WITH_EXCEPTION = 0;

	public static final byte RESPONSE_VALUE = 1;

	public static final byte RESPONSE_NULL_VALUE = 2;

	public static final Object[] EMPTY_OBJECT_ARRAY = new Object[0];

	public static final Class<?>[] EMPTY_CLASS_ARRAY = new Class<?>[0];

	static final byte MAGIC_HIGHT = (byte) 0xda;
	static final byte MAGIC_LOW = (byte) 0xbb;

	private Object service;

	private Class<?> serviceInterface;

	protected Interceptor interceptor;

	public DubboServletHandler() {
		this.interceptor = InterceptorUtils.getInterceptor();
	}

	/**
	 * Set the service to export. Typically populated via a bean reference.
	 */
	public void setService(Object service) {
		this.service = service;
	}

	/**
	 * Return the service to export.
	 */
	public Object getService() {
		return this.service;
	}

	/**
	 * Set the interface of the service to export. The interface must be
	 * suitable for the particular service and remoting strategy.
	 */
	public void setServiceInterface(Class<?> serviceInterface) {
		if (serviceInterface != null && !serviceInterface.isInterface()) {
			throw new IllegalArgumentException(
					"'serviceInterface' must be an interface");
		}
		this.serviceInterface = serviceInterface;
	}

	/**
	 * Return the interface of the service to export.
	 */
	public Class<?> getServiceInterface() {
		return this.serviceInterface;
	}

	InetAddress getRemoveAddress(HttpServletRequest request) {
		String ip = request.getHeader("x-forwarded-for");
		if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
			ip = request.getHeader("Proxy-Client-IP");
		}
		if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
			ip = request.getHeader("WL-Proxy-Client-IP");
		}
		if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
			ip = request.getRemoteAddr();
		}
		try {
			return InetAddress.getByName(ip);
		} catch (UnknownHostException e) {
			return null;
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public void handleRequest(HttpServletRequest request,
			HttpServletResponse response) throws ServletException, IOException {
		InetAddress remoteAddress = getRemoveAddress(request);
		if (this.interceptor != null
				&& !this.interceptor.acceptAfterConnected(remoteAddress)) {
			throw new IOException("deny!");
		}
		ServletInputStream input = request.getInputStream();
		byte[] header = new byte[20];
		if (input.read(header, 0, 16) != 16) {
			throw new IOException("Read invocation data failed.");
		}
		if (header[0] != MAGIC_HIGHT && header[1] != MAGIC_LOW)
			throw new IOException("Read invocation data failed.");
		byte flag = header[2];
		boolean native_dubbo = (flag & 0x10) == 0;
		byte proto = (byte) (flag & 0x0f);
		Serialization codec = CodecSupport.getSerializationById(proto);
		if (codec == null) {
			throw new IOException("Read invocation data failed.");
		}
		long id = Bytes.bytes2long(header, 4);
		int total_length = Bytes.bytes2int(header, 12);
		

		int data_length = total_length;
		int header_length = 0;
		if (!native_dubbo) {
			if (input.read(header, 16, 4) != 4) {
				throw new IOException("Read invocation data failed.");
			}
			header_length = Bytes.bytes2int(header, 16);
			data_length = total_length - header_length - 4;
		}
		
		int contentLength = request.getContentLength();
		if(contentLength != (total_length + 16)){
			throw new IOException("Read invocation data failed.");
		}
		if (this.interceptor != null
				&& !this.interceptor.acceptAfterHead(remoteAddress, total_length, data_length)) {
			throw new IOException("deny!");
		}
		final Map<String, Object> invoker = new HashMap<String, Object>();
		Map<String, String> headers = null;
		if (header_length > 0) {
			byte[] data = new byte[header_length];
			if (input.read(data, 0, data.length) != data.length)
				throw new IOException("Read invocation data failed.");
			InputStream is = new java.io.ByteArrayInputStream(data);
			decodeHeader(codec, is, invoker);
			headers = (Map<String, String>) invoker.get(HEADER_KEY);
			if (this.interceptor != null
					&& !this.interceptor.acceptAfterDecodeHeader(remoteAddress, headers, data_length)) {
				throw new IOException("deny!");
			}
			MessageDecode[] decodes = headers == null ? null
					: MessageCodecSupport.splitDecode(headers
							.get(MessageCodec.CODEC_KEY));
			//MessageEncodeRegistry registry = null;
			InputStream xis = input;
			if (decodes != null && decodes.length > 0) {
				//registry = new MessageEncodeRegistryImpl(headers);
				xis = new BridgeDecodeInputStream(new DecodeWrapper(decodes),
						input/*, registry*/);
			}
			decodeBody(codec, xis, invoker);
			if(xis != input){
				xis.close();
			}
			if(is != xis){
				is.close();
			}
		} else {
			decodeBody(codec, input, invoker);
		}
		
		if(this.interceptor != null){
			RemoteCalling calling = new RemoteCalling(){
				@Override
				public String target() {
					return (String)invoker.get(Constants.PATH_KEY);
				}

				@Override
				public String method() {
					return (String)invoker.get(METHOD_NAME_KEY);
				}

				@Override
				public Class<?>[] parameterTypes() {
					return ((Method)invoker.get(METHOD_KEY)).getParameterTypes();
				}

				@Override
				public Object[] parameters() {
					return (Object[])invoker.get(ARGS_KEY);
				}

				@Override
				public Map<String, String> headers() {
					return (Map<String, String>)invoker.get(HEADER_KEY);
				}
			};
		
			if(!this.interceptor.acceptAfterDecodeBody(remoteAddress, headers, calling)){
				throw new IOException("deny!");
			}
		}
		
		RpcContext context = RpcContext.getContext();
		Object result = null;
		try {
			Object target = invoker.get(METHOD_KEY);
			context.getRequestHeaders().setHeaders((Map<String, String>) invoker.get(HEADER_KEY));
			if(target instanceof HeaderProvider){
				((HeaderProvider)target).setHeaders(context.getRequestHeaders().getHeaders());
			}
			result = ((Method) invoker.get(METHOD_KEY)).invoke(service,
					(Object[]) invoker.get(ARGS_KEY));
			context = RpcContext.getContext();
			encodeResponse(request, response, codec, id, result,
					context.getResponseHeaders().getHeaders());
		} catch (Exception e) {
			encodeResponse(request, response, codec, id, e,
					context.getResponseHeaders().getHeaders());
		}

	}

	private static final String METHOD_NAME_KEY = "method-name";
	private static final String METHOD_KEY = "method";
	private static final String ARGS_KEY = "args";
	private static final String HEADER_KEY = "header";

	private void encodeResponse(HttpServletRequest request,
			HttpServletResponse response, Serialization codec, long id,
			Object result, Map<String, String> header) throws ServletException,
			IOException {
		UnsafeByteArrayOutputStream out = new UnsafeByteArrayOutputStream(4096);
		out.write(MAGIC_HIGHT);
		out.write(MAGIC_LOW);
		out.write(codec.getContentTypeId() | 0x10 | 0x40);
		out.write(0);

		out.writeInt64b(id); // 4-11
		out.writeInt32b(0); // 12-15
		out.writeInt32b(0); // 16-19

		ByteArrayOutputStream body = new ByteArrayOutputStream(4096);
		ByteArrayOutputStream hStream = new ByteArrayOutputStream(1024);
		int hStreamSize = 0;
		RpcContext context = RpcContext.getContext();
		Map<String, String> headers = context.getResponseHeaders().getHeaders();
		if (headers == null)
			headers = new HashMap<String, String>();
		
		if (headers != null && !headers.isEmpty()) {
			ObjectOutput o = codec.serialize(null, hStream);
			o.writeObject(headers);
			o.flushBuffer();
			hStreamSize = hStream.size();
			if (hStreamSize > 0) {
				hStream.writeTo(out);
			}
		}

		MessageEncode[] encodes = headers == null ? null : MessageCodecSupport
				.splitEncode(headers.get(MessageCodec.CODEC_KEY));

		if (encodes != null && encodes.length > 0) {
			//MessageEncodeRegistry registry = new MessageEncodeRegistryImpl(headers);
			OutputStream os = new BridgeEncodeOutputStream(new EncodeWrapper(
					encodes), body/*, registry*/);
			ObjectOutput o = codec.serialize(null, os);
			if (result == null) {
				o.writeInt(RESPONSE_NULL_VALUE);
			} else if (result instanceof Throwable) {
				o.writeInt(RESPONSE_WITH_EXCEPTION);
				o.writeObject(result);
			} else {
				o.writeInt(RESPONSE_VALUE);
				o.writeObject(result);
			}
			o.flushBuffer();
			os.close();
			/*
			if (headers != null) {
				headers.putAll(registry.getHeaders());
				headers.put(MessageCodec.CODEC_KEY,
						registry.getEncodesAsString());
			}
			*/
		} else {
			ObjectOutput o = codec.serialize(null, body);
			if (result == null) {
				o.writeInt(RESPONSE_NULL_VALUE);
			} else if (result instanceof Throwable) {
				o.writeInt(RESPONSE_WITH_EXCEPTION);
				o.writeObject(result);
			} else {
				o.writeInt(RESPONSE_VALUE);
				o.writeObject(result);
			}
			o.flushBuffer();
		}

		out.writeInt32b(12, body.size() + hStreamSize + 4);
		out.writeInt32b(16, hStreamSize);
		body.writeTo(out);
		
		response.setStatus(HttpServletResponse.SC_OK);
		response.setContentType("application/dubbo-octet-stream");
		response.setContentLength(out.size());
		
		out.writeTo(response.getOutputStream());
		
		body.close();
		hStream.close();
		out.close();
	}

	@SuppressWarnings("unchecked")
	private void decodeHeader(Serialization codec, InputStream input,
			Map<String, Object> invoker) throws IOException {
		ObjectInput in = codec.deserialize(null, input);
		try {
			Map<String, String> header = in.readObject(Map.class);
			if (header == null)
				header = new HashMap<String, String>();
			invoker.put(HEADER_KEY, header);
		} catch (ClassNotFoundException e) {
			throw new IOException(StringUtils.toString(
					"Read invocation data failed.", e));
		}
	}

	private void decodeBody(Serialization codec, InputStream input,
			Map<String, Object> invoker) throws IOException {
		ObjectInput in = codec.deserialize(null, input);
		invoker.put(Constants.DUBBO_VERSION_KEY, in.readUTF());
		String path = in.readUTF();
		invoker.put(Constants.PATH_KEY, path);
		invoker.put(Constants.VERSION_KEY, in.readUTF());

		String methodName = in.readUTF();
		invoker.put(METHOD_NAME_KEY, methodName);
		Method method = null;
		Object[] args;
		try {
			Class<?>[] pts = null;
			int paramCount = in.readInt();

			if (paramCount == 0) {
				pts = EMPTY_CLASS_ARRAY;
				args = EMPTY_OBJECT_ARRAY;
			} else if (paramCount < 0) {
				// aruan 2014.09.15 如果传入数据为-1
				// 说明客户端可能不是java,由服务器自行决定
				Class<?> target = Class.forName(path);
				Method m = ReflectUtils.findMethodByMethodName(target,
						methodName);
				if (m == null)
					throw new NoSuchMethodException(methodName);
				method = m;
				pts = m.getParameterTypes();
				paramCount = pts.length;
				args = new Object[pts.length];
				for (int i = 0; i < args.length; i++) {
					try {
						args[i] = in.readObject(pts[i]);
					} catch (Exception e) {
						if (log.isWarnEnabled()) {
							log.warn(
									"Decode argument failed: " + e.getMessage(),
									e);
						}
					}
				}
			} else {
				Class<?> target = Class.forName(path);
				Method m = ReflectUtils.findMethodByMethodName(target,
						methodName, paramCount);
				if (m == null)
					throw new NoSuchMethodException(methodName);
				method = m;
				pts = m.getParameterTypes();
				args = new Object[paramCount];
				for (int i = 0; i < args.length; i++) {
					try {
						args[i] = in.readObject(pts[i]);
					} catch (Exception e) {
						if (log.isWarnEnabled()) {
							log.warn(
									"Decode argument failed: " + e.getMessage(),
									e);
						}
					}
				}
			}

			ReflectUtils.makeCompatible(pts, args);
			// 翻译attachments
			in.readObject(Map.class);
			invoker.put(METHOD_KEY, method);
			invoker.put(ARGS_KEY, args);
			
		} catch (ClassNotFoundException e) {
			throw new IOException(StringUtils.toString(
					"Read invocation data failed.", e));
		} catch (NoSuchMethodException e) {
			throw new IOException(StringUtils.toString(
					"Read invocation data failed.", e));
		}
	}
}
